package regular;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class EvalReducer extends
		Reducer<IntWritable, RvecWritable, IntWritable, BooleanWritable> {

	public void reduce(IntWritable key, Iterable<RvecWritable> values,
			Context context) throws IOException, InterruptedException {

		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);

		// RvecWritable total = new RvecWritable();
		ArrayList<Rvec> RVset = new ArrayList<Rvec>();
		Iterator<RvecWritable> itr = values.iterator();
		RvecWritable data = new RvecWritable();
		while (itr.hasNext()) {
			data = itr.next();
		}
		RVset.addAll(data.rvset);

//		 for (Rvec r : data.rvset)
//		 System.out.println(r.toString());
		// System.out.println(gq.toString());
		// System.out.println(data.rvset.size());

		boolean answer = evalDG(RVset, gq);

		BooleanWritable val = new BooleanWritable(answer);
		// System.out.println(answer);
		context.write(key, val);
	}

	public GraphQuery getAutomaton() {
		// construct the query automaton
		VertexQ vq0 = new VertexQ("4");
		VertexQ vq1 = new VertexQ("DB");
		VertexQ vq2 = new VertexQ("HR");
		VertexQ vq3 = new VertexQ("12");

		vq0.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq2.stateName)));
		vq1.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq3.stateName)));
		vq2.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq2.stateName), new EdgeQ(vq3.stateName)));

		ArrayList<VertexQ> Vq = new ArrayList<VertexQ>(Arrays.asList(vq0, vq1,
				vq2, vq3));
		GraphQuery gq = new GraphQuery(Vq, vq0, vq3);
		return gq;
	}

	public GraphQuery getAutomaton1() {
		// construct the query automaton
		VertexQ vq0 = new VertexQ("2124");
		VertexQ vq1 = new VertexQ("Film & Animation");
		VertexQ vq2 = new VertexQ("Entertainment");
		VertexQ vq3 = new VertexQ("949");

		vq0.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq2.stateName)));
		vq1.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq3.stateName)));
		vq2.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq2.stateName), new EdgeQ(vq3.stateName)));

		ArrayList<VertexQ> Vq = new ArrayList<VertexQ>(Arrays.asList(vq0, vq1,
				vq2, vq3));
		GraphQuery gq = new GraphQuery(Vq, vq0, vq3);
		return gq;
	}

	// A BFS based function to check whether s can reach t
	public static boolean isReachable(ArrayList<VertexD> Vd, VertexD s,
			VertexD t) {
		if (s.equals(t))
			return true;

		// Create a queue for BFS
		Queue<VertexD> queue = new LinkedList<VertexD>();
		s.visit = true;
		queue.add(s);

		while (!queue.isEmpty()) {
			VertexD v = queue.poll();

			for (EdgeD e : v.neighbors) {
				// If this neighbor is the target vertex, then return true
				if (e.target.equals(t.name))
					return true;
				// Else, continue to do BFS
				int id = Vd.indexOf(new VertexD(e.target));				
				if (id >= 0) {
					if (Vd.get(id).visit == false) {
						queue.add(Vd.get(id));
						Vd.get(id).visit = true;
					}
				}
			}
		}
		return false;
	}

	public static boolean evalDG(ArrayList<Rvec> RVset, GraphQuery gq) {
		boolean answer = false;
		
		// construct dependence graph from RVset
		ArrayList<VertexD> Vd = new ArrayList<VertexD>();

		// Terminal state
		VertexD vdt = new VertexD(gq.ut.stateName, gq.ut.stateName);
		VertexD vds = new VertexD(gq.us.stateName, gq.us.stateName);

		boolean checkVdt = false;	
		// set neighbors
		for (Rvec r : RVset) {
			VertexD v = new VertexD(r.v, r.u);
			for (BoolFormula f : r.value) {
				if (f.value == true) {
					if (!v.name.equals(vdt.name)) // final state in in-node
						v.neighbors.add(new EdgeD(vdt.name));
					else
						checkVdt = true;
				} else if (f.boolVar != null) {
					v.neighbors.add(new EdgeD(f.boolVar.v));
				}
			}
			Vd.add(v);
		}

		// Add terminal vertex
		if (!checkVdt)
			Vd.add(vdt);	
		
//		for(VertexD v: Vd){
//			System.out.print(v.name+": ");
//			for(EdgeD e : v.neighbors)
//				System.out.print("\t"+e.target);
//			System.out.println();
//		}
		// get start vertex in Vq
		
		if (!Vd.contains(vds)) {
			System.out
					.println("Result is false because start state had not appeared in dependence graph!");
			return false;
		}
		else
		{
			int id = Vd.indexOf(vds);
			vds = Vd.get(id);
		}

		// BFS to get final answer
		answer = isReachable(Vd, vds, vdt);		

		return answer;
	}

	public static VertexD getVertexD(String name, ArrayList<VertexD> Vd) {
		VertexD tmp = null;
		for (VertexD v : Vd)
			if (v.name.equals(name)) {
				tmp = v;
				break;
			}

		return tmp;
	}

}
